package utils;

import com.alibaba.druid.pool.DruidDataSourceFactory;
import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;

import javax.sql.DataSource;
import java.sql.*;
import java.util.*;


/**
 * @ClassName ImpalaUtil
 * @Deacription 工具类
 * @Author wushumin
 * @Date 2020/4/26 8:51
 * @Version 1.0
 **/
@Slf4j
public class ImpalaUtil extends BaseDBManager {

    private static final Integer PAGESIZE = 10000;
    public static final String BLOOMFILTER_PREFIX = "mars-etl-userevent-bloomfilter:";
    public static final String USERID_FIRSTID_SPLIT = "|";
    public static final String USERID_FIRSTID_SPLIT_REG = "\\|";
    //    @Autowired
//    private static ApplicationContextUtils applicationContextUtils;

    private static DataSource impalaDataSource;
//    @Autowired
//    private DataSource dataPlatformImpalaDataSource;
//
//    @Autowired
//    private RedissonClient redissonClient;
//
//    @Autowired
//    private RedisTemplate<String, String> redisTemplate;
//
//    private HashOperations<String, String, String> opsForUserCache = null;
//    @PostConstruct
//    public void initImpala(){
//         opsForUserCache = redisTemplate.opsForHash();
//    }

    public static DataSource getImpalaDataSource(){
        Properties properties=new Properties();
        properties.put(DruidDataSourceFactory.PROP_DRIVERCLASSNAME,"com.cloudera.impala.jdbc41.Driver");
        //"jdbc:impala://192.168.6.76:21050/ods;UseNativeQuery=1"
        properties.put(DruidDataSourceFactory.PROP_URL, "jdbc:impala://192.168.6.76:21050/information_schema;UseNativeQuery=1");
//        properties.put(DruidDataSourceFactory.PROP_INITIALSIZE,1);
//        properties.put(DruidDataSourceFactory.PROP_MAXACTIVE,4);
//        properties.put(DruidDataSourceFactory.PROP_MAXWAIT,maxWait);
//        properties.put(DruidDataSourceFactory.PROP_TIMEBETWEENEVICTIONRUNSMILLIS,timeBetweenEvictionRunsMillis);
////        properties.put(DruidDataSourceFactory.PROP_MINEVICTABLEIDLETIMEMILLIS,minEvictableIdleTimeMillis);
        properties.put(DruidDataSourceFactory.PROP_VALIDATIONQUERY,"Select 1");
        properties.put(DruidDataSourceFactory.PROP_TESTWHILEIDLE,"true");
//        properties.put(DruidDataSourceFactory.PROP_TESTONBORROW,testOnBorrow);
//        properties.put(DruidDataSourceFactory.PROP_TESTONRETURN,testOnReturn);
//        properties.put(DruidDataSourceFactory.PROP_POOLPREPAREDSTATEMENTS,poolPreparedStatements);
        try {
            DataSource impalaDataSource = DruidDataSourceFactory.createDataSource(properties);
            log.info("======>create impala datasource:{}",impalaDataSource);
            return impalaDataSource;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    /**
     * 默认获取 sensor的impala data source -- this.impalaDataSource
     *
     * @return
     */
    public static Connection getConnection() {
        try {
//            DataSource dataSource= (DataSource) applicationContextUtils.getBean("impalaDataSource");
//            Assert.notNull(dataSource,"Impala datasource is null");
            Connection connection = getConnection(impalaDataSource);
            return connection;
        } catch (Exception e) {
            e.printStackTrace();
            log.info("======>get sensors connection error, datasource:{}", JSON.toJSONString(impalaDataSource), e);
        }
        return null;
    }

    public static Connection getConnection(DataSource dataSource) {
        try {
            Connection connection = dataSource.getConnection();
            return connection;
        } catch (SQLException e) {
            log.info("======>get sensors connection error, datasource:{}", JSON.toJSONString(dataSource), e);
        }
        return null;
    }

    /**
     * 普通查询语句
     *
     * @param sql
     * @param params
     */
    public static List<Map<String, Object>> executeQuery(String sql, List<Object> params) {
        return executeQuery(impalaDataSource, sql, params);
    }


    public static List<Map<String, Object>> executeQuery(DataSource dataSource, String sql, List<Object> params) {
        long begintime = System.currentTimeMillis();
        Connection conn = null;
        if (dataSource == null) {
            conn = getConnection();
        } else {
            conn = getConnection(dataSource);
        }
        if (conn == null) {
            log.error("======>get connectio error:{}");
            return null;
        }

        log.debug("======>sensorsCon:{}", conn);
        long connectionTime = System.currentTimeMillis();
        PreparedStatement pstmt = null;
        ResultSet rs = null;
        try {
            pstmt = conn.prepareStatement(sql);
            //为变量赋值
            if (CollectionUtils.isNotEmpty(params)) {
                for (int i = 1; i <= params.size(); i++) {
                    pstmt.setObject(i, params.get(i - 1));
                }
            }

            //执行查询
            rs = pstmt.executeQuery();
            //解析结果
            List<Map<String, Object>> resultList = pareseResult(rs);
            return resultList;

        } catch (Exception e) {
            log.error("======>executeQuery error , sql: {}, paras:{}", sql, JSON.toJSONString(params), e);
        } finally {
            BaseDBManager.releaseResource(conn, pstmt, rs);
            long endTime = System.currentTimeMillis();
            log.warn("=====impla connection time: {} ,query time: {}, sql:{},paras:{}", connectionTime - begintime, endTime - begintime,sql, JSON.toJSONString(params));
        }
        return null;
    }

    /**
     * 执行单条update、delete、insert；
     *
     * @param sql
     * @param params,可以为null
     */
    public int executeDML(String sql, List<Object> params) {
        long begintime = System.currentTimeMillis();
        Connection conn = null;
        conn = getConnection();
        if (conn == null) {
            log.error("======>get connectio error:{}");
            return -1;
        }
//        log.info("======>sensorsCon:{}", conn);
        long connectionTime = System.currentTimeMillis();
        PreparedStatement pstmt = null;
        try {
            conn.setAutoCommit(false);
            pstmt = conn.prepareStatement(sql);
            //为变量赋值
            if (CollectionUtils.isNotEmpty(params)) {
                ImpalaUtil.fillParams(pstmt, params);
            }
            //执行SQL
            int result = pstmt.executeUpdate();
            conn.commit();
            conn.setAutoCommit(true);
            return result;

        } catch (Exception e) {
            log.error("======>executeDML error , sql: {}, paras:{}", sql, JSON.toJSONString(params), e);
        } finally {
            BaseDBManager.releaseResource(conn, pstmt, null);
            long endTime = System.currentTimeMillis();
            log.info("=====executeDML get connection time: {} ,query time: {}", connectionTime - begintime, endTime - begintime);
        }
        return -1;
    }

    /**
     * 执行单条update、delete、insert；
     *
     * @param sql
     * @param params,可以为null
     */
    public boolean executeCommand(String sql, List<Object> params) {
        long begintime = System.currentTimeMillis();
        Connection conn = null;
        conn = this.getConnection();
        if (conn == null) {
            log.error("======>get connectio error:{}");
            return false;
        }
//        log.info("======>sensorsCon:{}", conn);
        long connectionTime = System.currentTimeMillis();
        PreparedStatement pstmt = null;
        try {
            conn.setAutoCommit(false);
            pstmt = conn.prepareStatement(sql);
            //为变量赋值
            if (CollectionUtils.isNotEmpty(params)) {
                ImpalaUtil.fillParams(pstmt, params);
            }
            //执行SQL
            boolean result = pstmt.execute();
            conn.commit();
            conn.setAutoCommit(true);
            return result;

        } catch (Exception e) {
            log.error("======>executeCommand error , sql: {}, paras:{}", sql, JSON.toJSONString(params), e);
        } finally {
            BaseDBManager.releaseResource(conn, pstmt, null);
            long endTime = System.currentTimeMillis();
            log.info("=====impla connection time: {} ,query time: {}", connectionTime - begintime, endTime - begintime);
        }
        return false;
    }


    /**
     * 封装结果集：key=columnName，value=columnValue
     *
     * @param rs
     * @return
     */
    public static List<Map<String, Object>> pareseResult(ResultSet rs) {
        if (rs != null) {
            List<Map<String, Object>> resultList = new ArrayList<Map<String, Object>>();

            List<String> columnList = new ArrayList<String>();
            try {
                ResultSetMetaData metaData = rs.getMetaData();
                int columnCount = metaData.getColumnCount();
                for (int i = 1; i <= columnCount; i++) {
                    columnList.add(metaData.getColumnName(i).toLowerCase());
                }

                while (rs.next()) {
                    Map<String, Object> rowMap = new HashMap<String, Object>(columnCount);
                    for (int i = 1; i <= columnCount; i++) {
                        rowMap.put(columnList.get(i - 1), rs.getObject(i));
                    }
                    resultList.add(rowMap);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
            return resultList;
        }
        return null;
    }

}
